//package com.example.rpc;
//
//
//import java.util.ArrayList;
//import java.util.List;
//import java.util.concurrent.CountDownLatch;
//import java.util.concurrent.Executor;
//import java.util.concurrent.Executors;
//
//import org.junit.After;
//import org.junit.Assert;
//import org.junit.Before;
//import org.junit.Test;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import com.alipay.remoting.Connection;
//import com.alipay.remoting.ConnectionEventType;
//import com.alipay.remoting.InvokeCallback;
//import com.alipay.remoting.exception.RemotingException;
//import com.alipay.remoting.rpc.RpcClient;
//import com.alipay.remoting.rpc.RpcResponseFuture;
//import com.alipay.remoting.rpc.common.BoltServer;
//import com.alipay.remoting.rpc.common.CONNECTEventProcessor;
//import com.alipay.remoting.rpc.common.DISCONNECTEventProcessor;
//import com.alipay.remoting.rpc.common.PortScan;
//import com.alipay.remoting.rpc.common.RequestBody;
//import com.alipay.remoting.rpc.common.SimpleClientUserProcessor;
//import com.alipay.remoting.rpc.common.SimpleServerUserProcessor;
//import com.alipay.remoting.util.RemotingUtil;
//
///**
// * basic usage demo
// *
// * basic usage of rpc client and rpc server
// *
// * @author xiaomin.cxm
// * @version $Id: BasicUsageDemo.java, v 0.1 Apr 6, 2016 8:58:36 PM xiaomin.cxm Exp $
// */
//public class BasicUsageDemoByJunit {
//    static Logger             logger                    = LoggerFactory
//                                                            .getLogger(BasicUsageDemoByJunit.class);
//
//    BoltServer                server;
//    RpcClient                 client;
//
//    int                       port                      = PortScan.select();
//    String                    ip                        = "127.0.0.1";
//    String                    addr                      = "127.0.0.1:" + port;
//
//    int                       invokeTimes               = 5;
//
//    SimpleServerUserProcessor serverUserProcessor       = new SimpleServerUserProcessor();
//    SimpleClientUserProcessor clientUserProcessor       = new SimpleClientUserProcessor();
//    CONNECTEventProcessor     clientConnectProcessor    = new CONNECTEventProcessor();
//    CONNECTEventProcessor     serverConnectProcessor    = new CONNECTEventProcessor();
//    DISCONNECTEventProcessor  clientDisConnectProcessor = new DISCONNECTEventProcessor();
//    DISCONNECTEventProcessor  serverDisConnectProcessor = new DISCONNECTEventProcessor();
//
//    @Before
//    public void init() {
//        server = new BoltServer(port, true);
//        server.start();
//        server.addConnectionEventProcessor(ConnectionEventType.CONNECT, serverConnectProcessor);
//        server.addConnectionEventProcessor(ConnectionEventType.CLOSE, serverDisConnectProcessor);
//        server.registerUserProcessor(serverUserProcessor);
//
//        client = new RpcClient();
//        client.addConnectionEventProcessor(ConnectionEventType.CONNECT, clientConnectProcessor);
//        client.addConnectionEventProcessor(ConnectionEventType.CLOSE, clientDisConnectProcessor);
//        client.registerUserProcessor(clientUserProcessor);
//        client.init();
//    }
//
//    @After
//    public void stop() {
//        try {
//            server.stop();
//            Thread.sleep(100);
//        } catch (InterruptedException e) {
//            logger.error("Stop server failed!", e);
//        }
//    }
//
//    /**
//     * <pre>
//     *     当前线程发起调用后，不关心调用结果，不做超时控制，只要请求已经发出，就完成本次调用。
//     *     注意 Oneway 调用不保证成功，而且发起方无法知道调用结果。因此通常用于可以重试，或者
//     *     定时通知类的场景，调用过程是有可能因为网络问题，机器故障等原因，导致请求失败。业务
//     *     场景需要能接受这样的异常场景，才可以使用
//     * </pre>
//     * @throws InterruptedException
//     */
//    @Test
//    public void testOneway() throws InterruptedException {
//        RequestBody req = new RequestBody(2, "hello world oneway");
//        for (int i = 0; i < invokeTimes; i++) {
//            try {
//                client.oneway(addr, req);
//                Thread.sleep(100);
//            } catch (RemotingException e) {
//                String errMsg = "RemotingException caught in oneway!";
//                logger.error(errMsg, e);
//                Assert.fail(errMsg);
//            }
//        }
//
//        Assert.assertTrue(serverConnectProcessor.isConnected());
//        Assert.assertEquals(1, serverConnectProcessor.getConnectTimes());
//        Assert.assertEquals(invokeTimes, serverUserProcessor.getInvokeTimes());
//    }
//
//    /**
//     * 同步调用
//     * <pre>
//     *     当前线程发起调用后，需要在指定的超时时间内，等到响应结果，才能完成本次调用
//     *     。如果超时时间内没有得到结果，那么会抛出超时异常。这种调用模式最常用。注意
//     *     要根据对端的处理能力，合理设置超时时间。
//     * </pre>
//     * @throws InterruptedException
//     */
//    @Test
//    public void testSync() throws InterruptedException {
//        RequestBody req = new RequestBody(1, "hello world sync");
//        for (int i = 0; i < invokeTimes; i++) {
//            try {
//                String res = (String) client.invokeSync(addr, req, 3000);
//                logger.warn("Result received in sync: " + res);
//                Assert.assertEquals(RequestBody.DEFAULT_SERVER_RETURN_STR, res);
//            } catch (RemotingException e) {
//                String errMsg = "RemotingException caught in sync!";
//                logger.error(errMsg, e);
//                Assert.fail(errMsg);
//            } catch (InterruptedException e) {
//                String errMsg = "InterruptedException caught in sync!";
//                logger.error(errMsg, e);
//                Assert.fail(errMsg);
//            }
//        }
//
//        Assert.assertTrue(serverConnectProcessor.isConnected());
//        Assert.assertEquals(1, serverConnectProcessor.getConnectTimes());
//        Assert.assertEquals(invokeTimes, serverUserProcessor.getInvokeTimes());
//    }
//
//    /**
//     * Future调用
//     * <pre>
//     *     当前线程发起调用，得到一个 RpcResponseFuture 对象，当前线程可以继续执行下一
//     *     次调用。可以在任意时刻，使用 RpcResponseFuture 对象的 get() 方法来获取结果，
//     *     如果响应已经回来，此时就马上得到结果；如果响应没有回来，则会阻塞住当前线程，直
//     *     到响应回来，或者超时时间到
//     * </pre>
//     * @throws InterruptedException
//     */
//    @Test
//    public void testFuture() throws InterruptedException {
//        RequestBody req = new RequestBody(2, "hello world future");
//        for (int i = 0; i < invokeTimes; i++) {
//            try {
//                RpcResponseFuture future = client.invokeWithFuture(addr, req, 3000);
//                String res = (String) future.get();
//                Assert.assertEquals(RequestBody.DEFAULT_SERVER_RETURN_STR, res);
//            } catch (RemotingException e) {
//                String errMsg = "RemotingException caught in future!";
//                logger.error(errMsg, e);
//                Assert.fail(errMsg);
//            } catch (InterruptedException e) {
//                String errMsg = "InterruptedException caught in future!";
//                logger.error(errMsg, e);
//                Assert.fail(errMsg);
//            }
//        }
//
//        Assert.assertTrue(serverConnectProcessor.isConnected());
//        Assert.assertEquals(1, serverConnectProcessor.getConnectTimes());
//        Assert.assertEquals(invokeTimes, serverUserProcessor.getInvokeTimes());
//    }
//
//    /**
//     * Callback异步调用
//     * <pre>
//     *     当前线程发起调用，则本次调用马上结束，可以马上执行下一次调用。发起调用时需要注册
//     *     一个回调，该回调需要分配一个异步线程池。待响应回来后，会在回调的异步线程池，来执
//     *     行回调逻辑
//     * </pre>
//     * @throws InterruptedException
//     */
//    @Test
//    public void testCallback() throws InterruptedException {
//        RequestBody req = new RequestBody(1, "hello world callback");
//        final List<String> rets = new ArrayList<String>(1);
//        for (int i = 0; i < invokeTimes; i++) {
//            final CountDownLatch latch = new CountDownLatch(1);
//            try {
//                client.invokeWithCallback(addr, req, new InvokeCallback() {
//                    Executor executor = Executors.newCachedThreadPool();
//
//                    @Override
//                    public void onResponse(Object result) {
//                        logger.warn("Result received in callback: " + result);
//                        rets.add((String) result);
//                        latch.countDown();
//                    }
//
//                    @Override
//                    public void onException(Throwable e) {
//                        logger.error("Process exception in callback.", e);
//                        latch.countDown();
//                    }
//
//                    @Override
//                    public Executor getExecutor() {
//                        return executor;
//                    }
//
//                }, 1000);
//
//            } catch (RemotingException e) {
//                latch.countDown();
//                String errMsg = "RemotingException caught in callback!";
//                logger.error(errMsg, e);
//                Assert.fail(errMsg);
//            }
//            try {
//                latch.await();
//            } catch (InterruptedException e) {
//                String errMsg = "InterruptedException caught in callback!";
//                logger.error(errMsg, e);
//                Assert.fail(errMsg);
//            }
//            if (rets.size() == 0) {
//                Assert.fail("No result! Maybe exception caught!");
//            }
//            Assert.assertEquals(RequestBody.DEFAULT_SERVER_RETURN_STR, rets.get(0));
//            rets.clear();
//        }
//
//        Assert.assertTrue(serverConnectProcessor.isConnected());
//        Assert.assertEquals(1, serverConnectProcessor.getConnectTimes());
//        Assert.assertEquals(invokeTimes, serverUserProcessor.getInvokeTimes());
//    }
//
//    @Test
//    public void testServerSyncUsingConnection1() throws Exception {
//        for (int i = 0; i < invokeTimes; i++) {
//            RequestBody req1 = new RequestBody(1, RequestBody.DEFAULT_CLIENT_STR);
//            String serverres = (String) client.invokeSync(addr, req1, 1000);
//            Assert.assertEquals(serverres, RequestBody.DEFAULT_SERVER_RETURN_STR);
//
//            Assert.assertNotNull(serverConnectProcessor.getConnection());
//            Connection serverConn = serverConnectProcessor.getConnection();
//            RequestBody req = new RequestBody(1, RequestBody.DEFAULT_SERVER_STR);
//            String clientres = (String) server.getRpcServer().invokeSync(serverConn, req, 1000);
//            Assert.assertEquals(clientres, RequestBody.DEFAULT_CLIENT_RETURN_STR);
//        }
//
//        Assert.assertTrue(serverConnectProcessor.isConnected());
//        Assert.assertEquals(1, serverConnectProcessor.getConnectTimes());
//        Assert.assertEquals(invokeTimes, serverUserProcessor.getInvokeTimes());
//    }
//
//    /**
//     * 示例1：使用 Connection 对象的双工通信
//     * <pre>
//     *     注意使用 Connection 对象的双工通信，服务端需要通过事件监听处理器或者用户请求处理器，自己保存好 Connection 对象
//     * </pre>
//     * @throws Exception
//     */
//    @Test
//    public void testServerSyncUsingConnection() throws Exception {
//        Connection clientConn = client.createStandaloneConnection(ip, port, 1000);
//
//        for (int i = 0; i < invokeTimes; i++) {
//            RequestBody req1 = new RequestBody(1, RequestBody.DEFAULT_CLIENT_STR);
//            String serverres = (String) client.invokeSync(clientConn, req1, 1000);
//            Assert.assertEquals(serverres, RequestBody.DEFAULT_SERVER_RETURN_STR);
//
//            Assert.assertNotNull(serverConnectProcessor.getConnection());
//            Connection serverConn = serverConnectProcessor.getConnection();
//            RequestBody req = new RequestBody(1, RequestBody.DEFAULT_SERVER_STR);
//            String clientres = (String) server.getRpcServer().invokeSync(serverConn, req, 1000);
//            Assert.assertEquals(clientres, RequestBody.DEFAULT_CLIENT_RETURN_STR);
//        }
//
//        Assert.assertTrue(serverConnectProcessor.isConnected());
//        Assert.assertEquals(1, serverConnectProcessor.getConnectTimes());
//        Assert.assertEquals(invokeTimes, serverUserProcessor.getInvokeTimes());
//    }
//
//    /**
//     * 示例2：使用 Address 的双工通信
//     * <pre>
//     *     注意使用 Address 方式的双工通信，需要在初始化 RpcServer 时，打开 manageConnection 开关，表示服务端会根据客户端发起的建连，维护一份地址与连接的映射关系。默认不需要双工通信的时候，这个功能是关闭的
//     * </pre>
//     * @throws Exception
//     */
//    @Test
//    public void testServerSyncUsingAddress() throws Exception {
//        Connection clientConn = client.createStandaloneConnection(ip, port, 1000);
//        String remote = RemotingUtil.parseRemoteAddress(clientConn.getChannel());
//        String local = RemotingUtil.parseLocalAddress(clientConn.getChannel());
//        logger.warn("Client say local:" + local);
//        logger.warn("Client say remote:" + remote);
//
//        for (int i = 0; i < invokeTimes; i++) {
//            RequestBody req1 = new RequestBody(1, RequestBody.DEFAULT_CLIENT_STR);
//            String serverres = (String) client.invokeSync(clientConn, req1, 1000);
//            Assert.assertEquals(serverres, RequestBody.DEFAULT_SERVER_RETURN_STR);
//
//            Assert.assertNotNull(serverConnectProcessor.getConnection());
//            // only when client invoked, the remote address can be get by UserProcessor
//            // otherwise, please use ConnectionEventProcessor
//            String remoteAddr = serverUserProcessor.getRemoteAddr();
//            RequestBody req = new RequestBody(1, RequestBody.DEFAULT_SERVER_STR);
//            String clientres = (String) server.getRpcServer().invokeSync(remoteAddr, req, 1000);
//            Assert.assertEquals(clientres, RequestBody.DEFAULT_CLIENT_RETURN_STR);
//        }
//
//        Assert.assertTrue(serverConnectProcessor.isConnected());
//        Assert.assertEquals(1, serverConnectProcessor.getConnectTimes());
//        Assert.assertEquals(invokeTimes, serverUserProcessor.getInvokeTimes());
//    }
//}
